/* Copyright  2009-2010
   @Author
   Richard Changde Yin            e-mail yinchangde@hotmail.com

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; version 2 of the License.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */

/**
 *@filename: hashIndex.cc
 */
#include <server_includes.h>
#include <allocator.h>
#include <base.h>
#include <m_ctype.h>
#include <util/test.h>

#include <string.h>


#include <index.h>
#include <catalogtables.h>
#include <lock.h>
#include <debug.h>
#include <table.h>
#include <database.h>
#include <transaction.h>

#include <gettext.h>
#include <error.h>
#include <errmsg_print.h>

using namespace BackEnd ;


#define REC_KEY_CMP(HASHIDXINFOPTR)                                   \
  do{                                                                 \
    if ((HASHIDXINFOPTR)->null_bit)                                   \
    {                                                                 \
      if ((rec1[(HASHIDXINFOPTR)->null_pos]&(HASHIDXINFOPTR)->null_bit) !=   \
          (rec2[(HASHIDXINFOPTR)->null_pos]&(HASHIDXINFOPTR)->null_bit))     \
            return 1;                                                 \
      if (rec1[(HASHIDXINFOPTR)->null_pos] & (HASHIDXINFOPTR)->null_bit)      \
            goto SUBLIST;                                               \
    }                                                                 \
    if ((HASHIDXINFOPTR)->ha_type == HA_KEYTYPE_TEXT)                \
    {                                                                 \
      const CHARSET_INFO * const cs=(HASHIDXINFOPTR)->charset;         \
      uint32_t char_length1;                                          \
      uint32_t char_length2;                                           \
      unsigned char *pos1= (unsigned char*)rec1 + (HASHIDXINFOPTR)->fldOffset;\
      unsigned char *pos2= (unsigned char*)rec2 + (HASHIDXINFOPTR)->fldOffset;\
      if (cs->mbmaxlen > 1)                                            \
      {                                                                \
        uint32_t char_length= (HASHIDXINFOPTR)->length / cs->mbmaxlen; \
        char_length1= my_charpos(cs, pos1, pos1+ (HASHIDXINFOPTR)->length,    \
                                 char_length);                         \
        set_if_smaller(char_length1, (uint32_t)(HASHIDXINFOPTR)->length);     \
        char_length2= my_charpos(cs, pos2, pos2+ (HASHIDXINFOPTR)->length,    \
                                 char_length);                         \
        set_if_smaller(char_length2, (uint32_t)(HASHIDXINFOPTR)->length);     \
      }                                                                \
      else                                                             \
      {                                                                \
        char_length1= char_length2= (HASHIDXINFOPTR)->length;          \
      }                                                                \
      if ((HASHIDXINFOPTR)->charset->coll->strnncollsp(cs,   \
                             pos1,char_length1,pos2,char_length2, 0))  \
        return 1;                                                      \
    }         /* Any VARCHAR segments */                               \
    else if ((HASHIDXINFOPTR)->ha_type == HA_KEYTYPE_VARTEXT1) /*TODO*/          \
    {                                                                        \
      unsigned char *pos1= (unsigned char*) rec1 + (HASHIDXINFOPTR)->fldOffset;\
      unsigned char *pos2= (unsigned char*) rec2 + (HASHIDXINFOPTR)->fldOffset;\
      uint32_t char_length1, char_length2;                                   \
      uint32_t pack_length= (HASHIDXINFOPTR)->bit_start;                    \
      const CHARSET_INFO * const cs= (HASHIDXINFOPTR)->charset;             \
      if (pack_length == 1)                                                  \
      {                                                                      \
        char_length1= (uint) *(unsigned char*) pos1++;                       \
        char_length2= (uint) *(unsigned char*) pos2++;                       \
      }                                                                      \
      else                                                                   \
      {                                                                      \
        char_length1= uint2korr(pos1);                                       \
        char_length2= uint2korr(pos2);                                        \
        pos1+= 2;                                                             \
        pos2+= 2;                                                             \
      }                                                                       \
      if (cs->mbmaxlen > 1)                                                   \
      {                                                                       \
        uint32_t safe_length1= char_length1;                                  \
        uint32_t safe_length2= char_length2;                                  \
        uint32_t char_length= (HASHIDXINFOPTR)->length/ cs->mbmaxlen;          \
        char_length1= my_charpos(cs, pos1, pos1 + char_length1, char_length); \
        set_if_smaller(char_length1, safe_length1);                           \
        char_length2= my_charpos(cs, pos2, pos2 + char_length2, char_length); \
        set_if_smaller(char_length2, safe_length2);                           \
      }                                                                       \
      if (cs->coll->strnncollsp((HASHIDXINFOPTR)->charset,                    \
                                pos1, char_length1,                          \
                                pos2, char_length2,                          \
                                (HASHIDXINFOPTR)->flag & HA_END_SPACE_ARE_EQUAL ? \
                                0 : diff_if_only_endspace_difference))        \
      return 1;                                                               \
    }                                                                         \
    else                                                                      \
    {                                                                         \
      if (memcmp(rec1+(HASHIDXINFOPTR)->start,rec2+(HASHIDXINFOPTR)->start,   \
                 (HASHIDXINFOPTR)->length))                                   \
      return 1;                                                               \
    }                                                                         \
  }while(0) ;

/*
  Compare keys for two records. Returns 0 if they are identical

  SYNOPSIS
    rec_key_cmp()
    HashIndexInfo
    rec1            Record to compare
    rec2            Other record to compare
    diff_if_only_endspace_difference
                    Different number of end space is significant

  NOTES
    diff_if_only_endspace_difference is used to allow us to insert
    'a' and 'a ' when there is an an unique key.

  RETURN
    0       Key is identical
    <> 0    Key is different
*/

int HashIndex::rec_key_cmp(HashIndexInfo *info,
                           const unsigned char *rec1,
                           const unsigned char *rec2,
                           bool diff_if_only_endspace_difference)
{
  HashIndexInfo* _hidxinfo = info ;
  BackEnd::ListIterator iter = _hidxinfo->sublist.getIterator();

  REC_KEY_CMP(_hidxinfo) ;

SUBLIST:
  while ( iter.hasElement()){
      _hidxinfo = (HashIndexInfo*) iter.nextElement() ;
      REC_KEY_CMP(_hidxinfo) ;
  }
  return 0;
}


   /* Calc hash value for a key in a record */

#define  REC_HASHNR(HIDXINFOPRT)                                             \
  do {                                                                       \
    pos =(unsigned char*) rec+(HIDXINFOPRT)->fldOffset;                      \
    if (info->null_bit)                                                      \
    {                                                                        \
      if (rec[(HIDXINFOPRT)->null_pos] & (HIDXINFOPRT)->null_bit)            \
      {                                                                      \
            nr^= (nr << 1) | 1;                                              \
            goto SUBLIST ;                                                   \
      }                                                                      \
    }                                                                        \
    if ((HIDXINFOPRT)->ha_type == HA_KEYTYPE_VARTEXT1                        \
        ||(HIDXINFOPRT)->ha_type == HA_KEYTYPE_VARTEXT2)  /* Any VARCHAR segments */    \
    {                                                                        \
      uint32_t pack_length= (HIDXINFOPRT)->bit_start;                        \
      uint32_t length= (pack_length == 1 ?                                   \
           (uint) *(unsigned char*) pos : uint2korr(pos));                  \
      info->charset->coll->hash_sort((HIDXINFOPRT)->charset, pos+pack_length,\
                                    length, &nr, &nr2);                     \
    }                                                                       \
    else                                                                    \
    {                                                                       \
      unsigned char *end= pos+(HIDXINFOPRT)->length;                        \
      for ( ; pos < end ; pos++)                                            \
      {                                                                     \
        nr *=16777619;                                                      \
        nr ^=(uint) *pos;                                                   \
      }                                                                     \
    }                                                                       \
  }while (0)

uint32_t HashIndex::cal_rec_key_hash(unsigned char* start,uint32_t len)
{
  uint32_t nr= 1;
  unsigned char *end= start+len;
  for (unsigned char* pos=start ; pos < end ; pos++)
  {
    nr *=16777619;
    nr ^=(uint) *pos;
  }
  return nr ;
}

uint32_t HashIndex::rec_hashnr(HashIndexInfo *info,
                               register const unsigned char *rec)
{
  uint32_t nr= 1, nr2= 4;

  HashIndexInfo* _hidxinfo = info ;
  unsigned char *pos  ;
  BackEnd::ListIterator iter = _hidxinfo->sublist.getIterator();
  REC_HASHNR(_hidxinfo) ;

SUBLIST:

  while ( iter.hasElement()){
      _hidxinfo = (HashIndexInfo*) iter.nextElement() ;
      REC_HASHNR(_hidxinfo) ;
  }
  return(nr);
}


#ifdef HAVE_ISNAN
#define CAL_RB_KEY(INDEXINFO)                                       \
                       do {                                           \
                          uint32_t char_length;                     \
                          if((INDEXINFO)->null_bit){           \
                            if (!(*key++= 1 - test(rec[(INDEXINFO)->null_pos] & (INDEXINFO)->null_bit)))\
                              goto RB_SUBLIST;\
                          }  \
                          if((INDEXINFO)->ha_type & HA_SWAP_KEY)\
                          {\
                            uint32_t length= (INDEXINFO)->length;\
                            unsigned char *pos= (unsigned char*) rec + (INDEXINFO)->start;\
                                  if ((INDEXINFO)->ha_type == HA_KEYTYPE_FLOAT)\
                                {\
                                     float nr;\
                                     float4get(nr, pos);\
                                     if (isnan(nr))\
                                    {\
                                     memset(key, 0, length);\
                                     key+= length;\
                                     goto RB_SUBLIST;\
                                    }\
                                }\
                                  else if ((INDEXINFO)->ha_type == HA_KEYTYPE_DOUBLE)\
                                {\
                                   double nr;\
                                   float8get(nr, pos);\
                                   if (isnan(nr))\
                                  {\
                                    memset(key, 0, length);\
                                    key+= length;\
                                    goto RB_SUBLIST;\
                                  }\
                                }\
                             pos+= length;\
                             while (length--)\
                            {\
                              *key++= *--pos;\
                            }\
                            goto RB_SUBLIST;\
                          }\
                          if ((INDEXINFO)->ha_type & HA_VAR_LENGTH_PART)\
                          {\
                            unsigned char *pos=   (unsigned char*) rec + (INDEXINFO)->start;\
                            uint32_t length=      (INDEXINFO)->length;\
                            uint32_t pack_length= (INDEXINFO)->bit_start;\
                            uint32_t tmp_length= (pack_length == 1 ? (uint) *(unsigned char*) pos :\
                                              uint2korr(pos));\
                            const CHARSET_INFO * const cs= (INDEXINFO)->charset;\
                            char_length= length/cs->mbmaxlen;\
                            pos+= pack_length;/* Skip VARCHAR length */\
                            set_if_smaller(length,tmp_length);\
                            FIX_LENGTH(cs, pos, length, char_length);\
                            store_key_length_inc(key,char_length);\
                            memcpy(key,pos,(size_t) char_length);\
                            key+= char_length;\
                            goto RB_SUBLIST;\
                          }\
                          char_length= (INDEXINFO)->length;\
                           if ((INDEXINFO)->charset->mbmaxlen > 1)\
                         {\
                             char_length= my_charpos((INDEXINFO)->charset,\
                                    rec + (INDEXINFO)->start, rec + (INDEXINFO)->start + char_length,\
                                    char_length / (INDEXINFO)->charset->mbmaxlen);\
                             set_if_smaller(char_length, (uint32_t)(INDEXINFO)->length); /* QQ: ok to remove? */\
                             if (char_length < (INDEXINFO)->length)\
                             (INDEXINFO)->charset->cset->fill((INDEXINFO)->charset, (char*) key + char_length,\
                                         (INDEXINFO)->length - char_length, ' ');\
                         }\
                         memcpy(key, rec + (INDEXINFO)->start, (size_t) char_length);\
                         key+= (INDEXINFO)->length;\
                    }while(0)
#else
#define CAL_RB_KEY(INDEXINFO)                                       \
                       do {                                           \
                          uint32_t char_length;                     \
                          if((INDEXINFO)->null_bit){           \
                            if (!(*key++= 1 - test(rec[(INDEXINFO)->null_pos] & (INDEXINFO)->null_bit)))\
                              goto RB_SUBLIST;\
                          }  \
                          if((INDEXINFO)->ha_type & HA_SWAP_KEY)\
                          {\
                            uint32_t length= (INDEXINFO)->length;\
                            unsigned char *pos= (unsigned char*) rec + (INDEXINFO)->start;\
                             pos+= length;\
                             while (length--)\
                            {\
                              *key++= *--pos;\
                            }\
                            goto RB_SUBLIST;\
                          }\
                          if ((INDEXINFO)->ha_type & HA_VAR_LENGTH_PART)\
                          {\
                            unsigned char *pos=   (unsigned char*) rec + (INDEXINFO)->start;\
                            uint32_t length=      (INDEXINFO)->length;\
                            uint32_t pack_length= (INDEXINFO)->bit_start;\
                            uint32_t tmp_length= (pack_length == 1 ? (uint) *(unsigned char*) pos :\
                                              uint2korr(pos));\
                            const CHARSET_INFO * const cs= (INDEXINFO)->charset;\
                            char_length= length/cs->mbmaxlen;\
                            pos+= pack_length;/* Skip VARCHAR length */\
                            set_if_smaller(length,tmp_length);\
                            FIX_LENGTH(cs, pos, length, char_length);\
                            store_key_length_inc(key,char_length);\
                            memcpy(key,pos,(size_t) char_length);\
                            key+= char_length;\
                            goto RB_SUBLIST;\
                          }\
                          char_length= (INDEXINFO)->length;\
                           if ((INDEXINFO)->charset->mbmaxlen > 1)\
                         {\
                             char_length= my_charpos((INDEXINFO)->charset,\
                                    rec + (INDEXINFO)->start, rec + (INDEXINFO)->start + char_length,\
                                    char_length / (INDEXINFO)->charset->mbmaxlen);\
                             set_if_smaller(char_length, (uint32_t)(INDEXINFO)->length); /* QQ: ok to remove? */\
                             if (char_length < (INDEXINFO)->length)\
                             (INDEXINFO)->charset->cset->fill((INDEXINFO)->charset, (char*) key + char_length,\
                                         (INDEXINFO)->length - char_length, ' ');\
                         }\
                         memcpy(key, rec + (INDEXINFO)->start, (size_t) char_length);\
                         key+= (INDEXINFO)->length;\
                    }while(0)
#endif
uint32_t HashIndex::make_rb_key(IndexInfo *info, unsigned char *key,
                                 const unsigned char *rec)
{
  unsigned char *start_key= key;
  IndexInfo* _hidxinfo = info ;
  BackEnd::ListIterator iter = _hidxinfo->sublist.getIterator();

  CAL_RB_KEY(_hidxinfo) ;

RB_SUBLIST:

  while ( iter.hasElement()){
    _hidxinfo = (HashIndexInfo*) iter.nextElement() ;
    CAL_RB_KEY(_hidxinfo) ;
  }
  return (uint32_t) (key - start_key);
}


#define RB_PACK_KEY(INDEXINFO)                                       \
                         do {   if(keypart_map){                     \
                                uint32_t char_length;\
                                keypart_map>>= 1;\
                                if ((INDEXINFO)->null_bit)\
                                {\
                                  if (!(*key++= (char) 1 - *old++))\
                                  goto RB_PACK_KEY_SUBLIST;\
                                }\
                                if ((INDEXINFO)->flag & HA_SWAP_KEY)\
                                {\
                                  uint32_t length= (INDEXINFO)->length;\
                                  unsigned char *pos= (unsigned char*) old + length;\
                                  while (length--)\
                                  {\
                                    *key++= *--pos;\
                                  }\
                                  goto RB_PACK_KEY_SUBLIST;\
                                }\
                                if ((INDEXINFO)->flag & (HA_VAR_LENGTH_PART | HA_BLOB_PART))\
                                {\
                                  /* Length of key-part used with heap_rkey() always 2 */\
                                  uint32_t tmp_length=uint2korr(old);\
                                  uint32_t length= (INDEXINFO)->length;\
                                  const CHARSET_INFO * const cs= (INDEXINFO)->charset;\
                                  char_length= length/cs->mbmaxlen;\
                                  old+= 2;\
                                  set_if_smaller(length,tmp_length); /* Safety */\
                                  FIX_LENGTH(cs, old, length, char_length);\
                                  store_key_length_inc(key,char_length);\
                                  memcpy(key, old,(size_t) char_length);\
                                  key+= char_length;\
                                  goto RB_PACK_KEY_SUBLIST;\
                                }\
                                char_length= (INDEXINFO)->length;\
                                if ((INDEXINFO)->charset->mbmaxlen > 1)\
                                {\
                                  char_length= my_charpos((INDEXINFO)->charset, old, old+char_length,\
                                                          char_length / (INDEXINFO)->charset->mbmaxlen);\
                                  set_if_smaller(char_length, (uint32_t)(INDEXINFO)->length); /* QQ: ok to remove? */\
                                  if (char_length < (INDEXINFO)->length)\
                                     (INDEXINFO)->charset->cset->fill((INDEXINFO)->charset, (char*) key + char_length,\
                                                 (INDEXINFO)->length - char_length, ' ');\
                                }\
                                memcpy(key, old, (size_t) char_length);\
                                key+= (INDEXINFO)->length;\
                                goto RB_PACK_KEY_SUBLIST;\
                          }\
                         }while(0)



uint32_t HashIndex::rb_pack_key(IndexInfo *info, unsigned char *key,
                                           const unsigned char *old,
                                           key_part_map keypart_map)
{
  unsigned char *start_key= key;
  IndexInfo* _hidxinfo = info ;
  BackEnd::ListIterator iter = _hidxinfo->sublist.getIterator();

  RB_PACK_KEY(_hidxinfo);

RB_PACK_KEY_SUBLIST:
  old+= (_hidxinfo)->length;
  while ( iter.hasElement()){
    _hidxinfo = (HashIndexInfo*) iter.nextElement() ;
    RB_PACK_KEY(_hidxinfo) ;
  }
  return (uint) (key - start_key);
}


DbRetVal HashIndex::insert(Table *tbl,Transaction *tr,
                          void *indexPtr,
                          IndexInfo* indInfo,void *tuple,bool undoFlag)
{
    HashIndexInfo *info = (HashIndexInfo*) indInfo;
    CINDEX *iptr = (CINDEX*)indexPtr;
    DbRetVal rc  = OK;
    size_t noOfBuckets   = info->noOfBuckets;
    uint32_t offset      = info->fldOffset;
    void *keyPtr =(void*)((unsigned char*)tuple + (size_t)offset);

    ChunkIterator citer = CatalogTableINDEX::getIterator(indexPtr);
    Bucket* buckets = (Bucket*)citer.nextElement();

    unsigned char keybuff [MAX_KEY_LENGTH];
    bzero(keybuff,MAX_KEY_LENGTH);
    uint32_t _key_len=make_rb_key((IndexInfo*)info, keybuff, (unsigned char*)tuple);


    uint32_t bucketNo =(uint32_t)(cal_rec_key_hash(keybuff,_key_len)% noOfBuckets);
    errmsg_printf(ERRMSG_LVL_DBUG, _("[%s]: line [%d] Insert HashIndex Bucket No %d \n"),
                                      __FILE__,__LINE__, bucketNo);

    Bucket *bucket =  &(buckets[bucketNo]);

    BKT_TREE_ARG custom_arg;

    HashUndoLogInfo logInfo;
    if (undoFlag && tr) {
      logInfo.tblPtr_  = tbl;     /* table ptr */
      logInfo.bucket_  = bucket;  /* bucket ptr*/
      logInfo.tuple_   = tuple;   /* tuple ptr */
      logInfo.indexPtr_= indexPtr;/* index ptr */
      logInfo.keyPtr_  = keyPtr;  /*  key ptr  */
    }
    int ret = pthread_rwlock_wrlock(&bucket->rwlock);
    if (ret)
    {
      return (DbRetVal) ret;
    }

    if(info->isUnique){
      bucket->subtree.setFlag(TREE_NO_DUPS);
    }else{
      bucket->subtree.setFlag( 0);
    }
    custom_arg.search_flag= SEARCH_FIND | SEARCH_UPDATE;
    custom_arg.key_length= _key_len;
    custom_arg.dbptr = tbl->getDB();
    custom_arg.chunkptr = bucket->element_chunk;
    custom_arg.info  = (IndexInfo*)info;

    BUCKET_TREE_ELEMENT* the_element = bucket->subtree.tree_insert(
                        (void*)keybuff,custom_arg.key_length, &custom_arg,ret);
    if(!the_element)
    {
      pthread_rwlock_unlock(&bucket->rwlock);
      return (DbRetVal)ret;
    }
    DbRetVal rv = OK;
    HashIndexNode *newNode=
         (HashIndexNode*)(((Chunk*)iptr->hashNodeChunk_)->allocate(tbl->getDB(),&rv));
      if (!newNode)
    {
      db_error(ER_OUTOFMEMORY, MYF(0), sizeof(HashIndexNode));
      rc = (DbRetVal)ER_OUTOFMEMORY;
        goto INSERTIDXERROR  ;
    }
      bucket->subtree.hashNodechunk = iptr->hashNodeChunk_;
      newNode->ptrToKey_   = keyPtr;
      newNode->ptrToTuple_ = tuple;
      newNode->next_       = NULL;
     if (!the_element->bucketList_->getSize())
    {
      errmsg_printf(ERRMSG_LVL_DBUG, _("[%s]: line [%d] Record [%p]. HashIndex [%s] sublist size is ZERO.\n"),
                                        __FILE__,__LINE__,tuple,info->idxName );
    }
    the_element->bucketList_->insertNode((HashIndexNode*)newNode);
    errmsg_printf(ERRMSG_LVL_DBUG, _("[%s]: line [%d] HashIndex insert new Index Node [%p] into the bucket. No [%d].Tree element [%p].\n"),
                                                 __FILE__,__LINE__, newNode,bucketNo,the_element);
    if (undoFlag && tr) {
         rc = tr->appendLogicalHashUndoLog(tbl->getSystemDB(),
                                           InsertHashIndexOperation,
                                           &logInfo,
                                           sizeof(HashUndoLogInfo));
        if (rc !=OK)
      {
        the_element->bucketList_->removeNode((Chunk*)iptr->hashNodeChunk_, tbl->getDB(), newNode);
      }
    }
    if(!the_element->bucketList_->getSize()&&info->isUnique){
    // remove the tree element.
       bucket->subtree.tree_delete((void*)keybuff,custom_arg.key_length,&custom_arg);
    }
INSERTIDXERROR:

    pthread_rwlock_unlock(&bucket->rwlock);
    return rc;
}

DbRetVal HashIndex::remove(Table*      tbl,
                           Transaction *tr,
                           void *indexPtr ,
                           IndexInfo *indInfo,
                           void *tuple,
                           void *recpos,
                           bool undoFlag)
{
    if(!recpos) return(DbRetVal) HA_ERR_WRONG_IN_RECORD;
    HashIndexInfo *info = (HashIndexInfo*) indInfo;
    DbRetVal rc = OK;
    int noOfBuckets     = info->noOfBuckets;
    uint32_t offset     = info->fldOffset;
    void *keyPtr =(void*)((unsigned char*)recpos +(size_t) offset);

    ChunkIterator citer = CatalogTableINDEX::getIterator(indexPtr);
    Bucket* buckets = (Bucket*)citer.nextElement();
    unsigned char keybuff [MAX_KEY_LENGTH];
    bzero(keybuff,MAX_KEY_LENGTH);
    uint32_t _key_len=make_rb_key((IndexInfo*)info, keybuff, (unsigned char*)tuple);

    uint32_t bucketNo =(uint32_t)(cal_rec_key_hash(keybuff,_key_len)% noOfBuckets);
    CINDEX *iptr = (CINDEX*)indexPtr;


    Bucket *bucket = &buckets[bucketNo];
    HashUndoLogInfo hInfo ;
    hInfo.tblPtr_ = tbl;
    hInfo.bucket_ = bucket;
    hInfo.tuple_  = recpos;
    hInfo.indexPtr_ = indexPtr;
    // TODO... It is not correct.
    hInfo.keyPtr_ = keyPtr;
    int ret = pthread_rwlock_wrlock(&bucket->rwlock);
    if (ret)
       return (DbRetVal)ret;


    BKT_TREE_ARG custom_arg;


    if(info->isUnique){
         bucket->subtree.setFlag(TREE_NO_DUPS);
    }else{
         bucket->subtree.setFlag( 0);
    }
    custom_arg.search_flag= SEARCH_FIND | SEARCH_UPDATE;
    custom_arg.key_length= _key_len;
    custom_arg.dbptr = tbl->getDB();
    custom_arg.chunkptr = bucket->element_chunk;
    custom_arg.info  = (IndexInfo*)info;
    custom_arg.search_type = get_element;
    BUCKET_TREE_ELEMENT* the_element =
                        (BUCKET_TREE_ELEMENT*)bucket->subtree.tree_search((void*)keybuff,&custom_arg);

    if(!the_element){
       goto REMOVE_INDEX_LEAVE;
    }
    the_element->bucketList_->remove((Chunk*)iptr->hashNodeChunk_, tbl->getDB(), keyPtr);

    if (undoFlag && tr) {
        rc =tr->appendLogicalHashUndoLog(tbl->getSystemDB(),
                                         DeleteHashIndexOperation,
                                         &hInfo,
                                         sizeof(HashUndoLogInfo));
        if (rc !=OK)
        {
            rc = the_element->bucketList_->insert((Chunk*)iptr->hashNodeChunk_, tbl->getDB(),
                              keyPtr, recpos);
        }
    }
    if(!the_element->bucketList_->getSize()){
    // remove the tree element.
       bucket->subtree.tree_delete((void*)keybuff,custom_arg.key_length,&custom_arg);
    }

 REMOVE_INDEX_LEAVE:
    pthread_rwlock_unlock(&bucket->rwlock);
    return rc;
}


DbRetVal HashIndex::update (Table *,    //tbl,
                           Transaction *,//tr,
                           void *,//indexPtr,
                           IndexInfo *,//indInfo,
                           void *,//tuple,
                           bool //undoFlag
                           )
{

    return OK ;
    /*
    CINDEX *iptr = (CINDEX*)indexPtr;

    HashIndexInfo *info = (HashIndexInfo*) indInfo;
    int offset  = info->fldOffset;
    int noOfBuckets = info->noOfBuckets;

    //check whether the index key is updated or not
    //if it is not updated return from here
    void *keyPtr =(void*)((unsigned char*)tuple + offset);
    char *kPtr= (char*)keyPtr;

    //creating old key value buffer for composite primary keys
    char *oldKeyBuffer = (char*) malloc(info->compLength);
    void *oldKeyStartBuffer = oldKeyBuffer;
    FieldIterator iter = info->idxFldList.getIterator();
    while(iter.hasElement()) {
        FieldDef *def = iter.nextElement();
        keyPtr = (char *)tuple + def->offset_;
        AllDataType::copyVal(oldKeyBuffer, keyPtr, def->type_, def->length_);
        oldKeyBuffer = oldKeyBuffer + AllDataType::size(def->type_, def->length_);
    }

    keyPtr = (void *) kPtr;
    //Iterate through the bind list and check
    FieldIterator idxFldIter = info->idxFldList.getIterator();
    char *keyBindBuffer ;
    if(type==typeBinary) keyBindBuffer = (char*) malloc(2 * info->compLength);
    else keyBindBuffer = (char*) malloc(info->compLength);
    void *keyStartBuffer = (void*) keyBindBuffer;
    bool keyUpdated = false;

    while (idxFldIter.hasElement()) {
        FieldDef *idef = idxFldIter.nextElement();
        FieldIterator fldIter = tbl->fldList_.getIterator();
        while (fldIter.hasElement()) {
            FieldDef *def = fldIter.nextElement();
            if (0 == strcmp(def->fldName_, idef->fldName_)) {
                if (NULL != def->bindVal_) {
                    if(type==typeBinary) {
                        AllDataType::copyVal(keyBindBuffer, def->bindVal_,
                                                   def->type_, 2*def->length_);
                        keyStartBuffer=calloc(1,info->compLength);
                        AllDataType::convertToBinary(keyStartBuffer, keyBindBuffer, typeString, info->compLength);
                        free(keyBindBuffer);
                    } else {
                        AllDataType::copyVal(keyBindBuffer, def->bindVal_,
                                                   def->type_, def->length_);
                        keyBindBuffer = keyBindBuffer + AllDataType::size(def->type_, def->length_);
                    }
                } else {
                     AllDataType::copyVal(keyBindBuffer, (char *) tuple + def->offset_, def->type_, def->length_);
                     keyBindBuffer = keyBindBuffer + AllDataType::size(def->type_, def->length_);
                }
                keyUpdated = true;
                break;
            }
        }
    }
    if (!keyUpdated) {
        //printf("PRABA::key not updated\n");
        free(keyStartBuffer);
        free(oldKeyStartBuffer);
        return OK;
    }
    //printf("PRABA::it is wrong coming here\n");
    bool result = false;
    if (type == typeComposite)
        result = AllDataType::compareVal(oldKeyStartBuffer, keyStartBuffer,
                                OpEquals, info->type, info->compLength);
    else result = AllDataType::compareVal(keyPtr, keyStartBuffer,
                                OpEquals, info->type, info->compLength);
    if (result) {
        free(keyStartBuffer);
        free(oldKeyStartBuffer);
        return OK;
    }
    printDebug(DM_HashIndex, "Updating hash index node: Key value is updated");

    ChunkIterator citer = CatalogTableINDEX::getIterator(indexPtr);

    Bucket* buckets = (Bucket*)citer.nextElement();

    //remove the node whose key is updated
    int bucketNo = 0;
    if (type == typeComposite)
        bucketNo = computeHashBucket(type, oldKeyStartBuffer, noOfBuckets, info->compLength);
    else bucketNo = computeHashBucket(type, keyPtr, noOfBuckets, info->compLength);
    printDebug(DM_HashIndex, "Updating hash index node: Bucket for old value is %d", bucketNo);
    Bucket *bucket = &buckets[bucketNo];

    HashUndoLogInfo *hInfo1 = new HashUndoLogInfo();
    hInfo1->tblPtr_ = tbl;
    hInfo1->bucket_ = bucket;
    hInfo1->tuple_ = tuple;
    hInfo1->indexPtr_ = indexPtr;
    hInfo1->keyPtr_ = keyPtr;

    //it may run into deadlock, when two threads updates tuples which falls in
    //same buckets.So take both the mutex one after another, which will reduce the
    //deadlock window.
    int ret = bucket->mutex_.getLock(tbl->db_->procSlot);
    if (ret != 0)
    {
        delete hInfo1;
        free(keyStartBuffer);
        free(oldKeyStartBuffer);
        printError(ErrLockTimeOut,"Unable to acquire bucket Mutex for bucket %d",bucketNo);
        return ErrLockTimeOut;
    }
    //insert node for the updated key value
    int newBucketNo = computeHashBucket(type,
                        keyStartBuffer, noOfBuckets, info->compLength);
    printDebug(DM_HashIndex, "Updating hash index node: Bucket for new value is %d", newBucketNo);

    Bucket *bucket1 = &buckets[newBucketNo];
    HashUndoLogInfo *hInfo2 = new HashUndoLogInfo();
    hInfo2->tblPtr_ = tbl;
    hInfo2->bucket_ = bucket;
    hInfo2->tuple_ = tuple;
    hInfo2->indexPtr_ = indexPtr;
    hInfo2->keyPtr_ = keyPtr;
    bucket1->mutex_.getLock(tbl->db_->procSlot);
    if (ret != 0)
    {
        delete hInfo1;
        delete hInfo2;
        free(keyStartBuffer);
        free(oldKeyStartBuffer);
        printError(ErrLockTimeOut,"Unable to acquire bucket Mutex for bucket %d",newBucketNo);
        return ErrLockTimeOut;
    }

    HashIndexNode *head1 = (HashIndexNode*) bucket->bucketList_;
    if (head1)
    {
        BucketList list1(head1);
        printDebug(DM_HashIndex, "Updating hash index node: Removing node from list with head %x", head1);
        list1.remove((Chunk*)iptr->hashNodeChunk_, tbl->db_, keyPtr);
        bucket->bucketList_=list1.getBucketListHead();
    }
    else
    {
        printError(ErrSysInternal,"Update: Bucket list is null");
        bucket1->mutex_.releaseLock(tbl->db_->procSlot);
        bucket->mutex_.releaseLock(tbl->db_->procSlot);
        delete hInfo1;
        delete hInfo2;
        free(keyStartBuffer);
        free(oldKeyStartBuffer);
        return ErrSysInternal;
    }
    DbRetVal rc = OK;
    if (undoFlag) {
         rc = tr->appendLogicalHashUndoLog(tbl->sysDB_, DeleteHashIndexOperation, hInfo1, sizeof(HashUndoLogInfo));
        if (rc !=OK)
        {
            BucketList list((HashIndexNode*) bucket->bucketList_);
            rc = list.insert((Chunk*)iptr->hashNodeChunk_, tbl->db_, keyPtr, tuple);
            if (rc !=OK) printError(ErrSysFatal, "double failure on undo log remove followed by hash bucket list insert\n");
            bucket->bucketList_ = list.getBucketListHead();
            bucket1->mutex_.releaseLock(tbl->db_->procSlot);
            bucket->mutex_.releaseLock(tbl->db_->procSlot);
            delete hInfo1;
            delete hInfo2;
            free(keyStartBuffer);
            free(oldKeyStartBuffer);
            return rc;
        }
    }
    HashIndexNode *head2 = (HashIndexNode*) bucket1->bucketList_;
    //Note:: the tuple will be in the same address location
    //so not changing the keyptr and tuple during append
    //only bucket where this node resides will only change
    //if the index key is updated.
    if (!head2)
    {
        DbRetVal rv = OK;
        HashIndexNode *firstNode= (HashIndexNode*)(((Chunk*)iptr->hashNodeChunk_)->allocate(tbl->db_, &rv));
        if (firstNode == NULL)
        {
            printError(rv, "Error in allocating hash node");
            bucket1->mutex_.releaseLock(tbl->db_->procSlot);
            bucket->mutex_.releaseLock(tbl->db_->procSlot);
            delete hInfo1;
            delete hInfo2;
            free(keyStartBuffer);
            free(oldKeyStartBuffer);
            return rv;
        }
        firstNode->ptrToKey_ = keyPtr;
        firstNode->ptrToTuple_ = tuple;
        firstNode->next_ = NULL;
        bucket1->bucketList_ = (HashIndexNode*)firstNode;
        printDebug(DM_HashIndex, "Updating hash index node: Adding new node %x:Head is empty", firstNode);
    }
    else
    {
        BucketList list2(head2);
        printDebug(DM_HashIndex, "Updating hash index node: Adding node to list with head %x", head2);
        list2.insert((Chunk*)iptr->hashNodeChunk_, tbl->db_, keyPtr, tuple);
        bucket1->bucketList_ = list2.getBucketListHead();
    }
    if (undoFlag) {

        rc = tr->appendLogicalHashUndoLog(tbl->sysDB_, InsertHashIndexOperation, hInfo2, sizeof(HashUndoLogInfo));
        if (rc !=OK)
        {
            //reverting back the changes:delete new node and add the old
            //node + remove logical undo log of the  DeleteHashIndexOperation
            BucketList list1((HashIndexNode*) bucket->bucketList_);
            BucketList list2((HashIndexNode*) bucket1->bucketList_);
            list1.insert((Chunk*)iptr->hashNodeChunk_, tbl->db_, keyPtr, tuple);
            list2.remove((Chunk*)iptr->hashNodeChunk_, tbl->db_, keyPtr);
            bucket->bucketList_ = list1.getBucketListHead();
            bucket1->bucketList_ = list2.getBucketListHead();
            UndoLogInfo *logInfo = tr->popUndoLog();
            Chunk *chunk = tbl->sysDB_->getSystemDatabaseChunk(UndoLogTableID);
            chunk->free(tbl->sysDB_, logInfo);
        }
    }
    bucket1->mutex_.releaseLock(tbl->db_->procSlot);
    bucket->mutex_.releaseLock(tbl->db_->procSlot);
    delete hInfo1;
    delete hInfo2;
    free(keyStartBuffer);
    free(oldKeyStartBuffer);
    return rc;
    */
}

//Following methods are used to undo Logical Hash Indexes
DbRetVal HashIndex::insertLogicalUndoLog(Database *, void *data)
{
    HashUndoLogInfo *info = (HashUndoLogInfo *) data;
    Table* tbl    = (Table *) info->tblPtr_;
    Chunk* hChunk = (Chunk *) ((CINDEX *)info->indexPtr_)->hashNodeChunk_;

    HashIndexNode* head =0;
     //      (HashIndexNode*)((Bucket *)(info->bucket_))->bucketList_;
    BucketList list(head);
    list.insert(hChunk,
                tbl->getDB(),
                info->keyPtr_,
                info->tuple_);
    //((Bucket *)info->bucket_)->bucketList_ = list.getBucketListHead();

    return OK;
}

DbRetVal HashIndex::deleteLogicalUndoLog(Database *, void *data)
{
    HashUndoLogInfo* info = (HashUndoLogInfo *) data;
    Table * tbl = (Table*)info->tblPtr_;
    Chunk *hChunk = (Chunk *) ((CINDEX *)info->indexPtr_)->hashNodeChunk_;
    HashIndexNode *head=0;
              //    (HashIndexNode *)((Bucket *)info->bucket_)->bucketList_;
    BucketList list(head);
    list.remove(hChunk, tbl->getDB(), info->keyPtr_);
    //((Bucket *)info->bucket_)->bucketList_ = list.getBucketListHead();
    return OK;
}
